package com.study.chapter08;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/11 22:14
 */
public class ConnectStreamTest2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //app订单
        SingleOutputStreamOperator<Tuple3<String, String, Long>> appStream = env.fromElements(
                Tuple3.of("order-1", "app", 1000L),
                Tuple3.of("order-3", "app", 3000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<Tuple3<String, String, Long>>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner((tup, l) -> tup.f2)
        );

        //第3三支付平台
        SingleOutputStreamOperator<Tuple4<String, String,String, Long>> thirdPartStream = env.fromElements(
                Tuple4.of("order-1", "third","sucess", 1000L),
                Tuple4.of("order-2", "third","sucess", 1000L),
                Tuple4.of("order-3", "third", "sucess",3000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<Tuple4<String, String,String, Long>>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner((tup, l) -> tup.f3)
        );

        //检测同一支付单在两条流中是否匹配，不匹配就报警
        //写法一
        appStream.keyBy(data -> data.f0)
                .connect(thirdPartStream.keyBy(data -> data.f0));

        //写法二
        appStream.connect(thirdPartStream)
                .keyBy(app -> app.f0,third -> third.f0)
                .process(new OrderMatchProcess())
                .print();


        env.execute();
    }

    //自定义实现CoProcessFunction
    public static class OrderMatchProcess extends CoProcessFunction<Tuple3<String, String, Long>,Tuple4<String, String,String, Long>,String>{
        //定义状态变量，用来保存已经到达的事件
        private ValueState<Tuple3<String, String, Long>> appState;

        private ValueState<Tuple4<String, String, String, Long>> thirdPartState;

        @Override
        public void open(Configuration parameters) throws Exception {
            appState = getRuntimeContext().getState(
                    new ValueStateDescriptor<Tuple3<String, String, Long>>("app-event", Types.TUPLE(Types.STRING,Types.STRING,Types.LONG))
            );
            thirdPartState = getRuntimeContext().getState(
                    new ValueStateDescriptor<Tuple4<String, String, String,Long>>("thirdPart-event", Types.TUPLE(Types.STRING,Types.STRING,Types.STRING,Types.LONG))
            );
        }

        @Override
        public void processElement1(Tuple3<String, String, Long> value, Context ctx, Collector<String> out) throws Exception {
            //来的是app event，看另一条流中事件是否来过
            if(thirdPartState.value() != null){
                out.collect("对账成功："+value + "  "+thirdPartState.value());
                //清空状态
                thirdPartState.clear();
            }else {
                //更新状态
                appState.update(value);
                //注册一个5秒后的定时器，开始等待另一条流的事件
                ctx.timerService().registerEventTimeTimer(value.f2 + 5000L);
            }
        }

        @Override
        public void processElement2(Tuple4<String, String, String, Long> value, Context ctx, Collector<String> out) throws Exception {
                //来的是thirdPart event，看另一条流中事件是否来过
            if(appState.value() != null){
                out.collect("对账成功："+appState.value() + "  "+value);
                //清空状态
                appState.clear();
            }else {
                //更新状态
                thirdPartState.update(value);
                //注册一个5秒后的定时器，开始等待另一条流的事件
                ctx.timerService().registerEventTimeTimer(value.f3);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            if (appState.value() != null){
                out.collect("对账失败："+appState.value() + "  "+"第三方信息未到");
            }
            if (thirdPartState.value() != null){
                out.collect("对账失败："+thirdPartState.value() + "  "+"app信息未到");
            }
            appState.clear();
            thirdPartState.clear();
        }
    }
}
